package com.ams.server.replication;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ams.io.network.connection.Connection;
import com.ams.protocol.rtmp.RtmpConnection;
import com.ams.protocol.rtmp.RtmpHandShake;
import com.ams.server.handler.ProtocolHandler;

public class ReplPublishHandler extends ProtocolHandler {
    final private Logger logger = LoggerFactory
            .getLogger(ReplPublishHandler.class);

    private RtmpHandShake handshake;
    private ReplPublisher publisher;
    private boolean keepAlive = true;

    public void run() {
        if (conn.isClosed()) {
            close();
            return;
        }
        try {
            // wait until server handshake done
            if (!handshake.isHandshakeDone()) {
                handshake.doServerHandshake();
                if (handshake.isHandshakeDone()) {
                    logger.info("replication rtmp handshake done.");
                }
                conn.flush();
                return;
            }
            // publish stream to subscriber
            publisher.acceptSubscription();

            publisher.tryToCloseSubscribeStreams();

            publisher.pingSubscriber();

            Thread.sleep(10);

            conn.flush();
        } catch (Exception e) {
            logger.debug(e.getMessage());
            close();
        }
    }

    public void close() {
        conn.close();
        keepAlive = false;
    }

    public void setConnection(Connection conn) {
        this.conn = conn;
        RtmpConnection rtmp = new RtmpConnection(conn);
        this.publisher = new ReplPublisher(rtmp);
        this.handshake = new RtmpHandShake(rtmp);
    }

    public boolean isKeepAlive() {
        return keepAlive;
    }

    public ProtocolHandler handlerFactory() {
        return new ReplPublishHandler();
    }

}